#include "util.h"
// #include "circularbuffer.h"
#include "sync.h"
#include <fstream>
#include <sstream>
#include <stdarg.h>

#define OPENBRACE {
#define CLOSEBRACE }

// An Sampler samples from tuples which start with any given prefix.  
// Tuple: fixed width tuple with a weight
// Prefix: represented in a Tuple, but has a prefix_length__ instead of a weight_
//         Elements past the length of the prefix are ignored.

namespace Ads OPENBRACE

enum AD_FIELD {
  LHS,
  RHS,
  AD,
  NUM_FIELDS,
};

const uint64 WIDTH = NUM_FIELDS;

struct Tuple {
  uint64 a_[WIDTH];
  
  // Before an Sampler is built, weight_ represents the actual weight
  // on the tuple. 
  // After the Sampler is built, weight_ represents the sum of all 
  // weights up to and including this tuple.  This is useful for sampling. 
  uint64 weight_;
  
  // We also use the Tuple structure to represent prefixes. 
  // In that case, we use the weight_ field to represent the prefix length. 
#define prefix_length__ weight_
  
  // Tuples compare lexicographically.
  bool operator <(const Tuple & o) const {
    for (uint64 i=0; i<WIDTH; i++) {
      if (a_[i] < o.a_[i]) return true;
      if (a_[i] > o.a_[i]) return false;
    } return false; }
  
  // For debugging. 
  string ToString() const;
  
  // for debugging - reads the weight then the tuple from an input stream
  bool Read(istream &input);
  
  // for debugging - reads the length then the prefix from an input stream
  bool ReadPrefix(istream &input);
  
  // The set of tuples with a particular prefix is [lower_bound upper_bound) .
  // Here we compute those bounds.
  Tuple PrefixToLowerBound() const;
  
  Tuple PrefixToUpperBound() const;

  uint64 NodeCount() const;
  uint64 WeightPerNode() const;
};

// Create a prefix from variable args.
Tuple Prefix(uint64 prefix_length, ...);

// create an entire tuple
Tuple FullTuple(uint64 weight, ...);
 
// Compare tuples by weight - useful for sampling. 
struct WeightCompare {
  bool operator() (const Tuple & r1, const Tuple & r2) const { 
    return (r1.weight_ < r2.weight_); } 
};

enum SampleStrategy {
  SAMPLE_WEIGHTED,
  SAMPLE_UNIFORM_OVER_SAMPLERS,
  SAMPLE_UNIFORM_OVER_ADS,
  SAMPLE_NUM_STRATEGIES,
};

struct Sampler {
  // Sampler data
  uint64 size_;
  uint64 capacity_;
  Tuple * data_;
  bool built_;
  
  Sampler() { size_ = NULL; data_ = NULL; } 
  ~Sampler();
  bool IsFull() { return (size_ == capacity_); }
  string ToString();
  void Allocate(uint64 capacity);
  // void ReadData(CircularBuffer * buffer);
  void AddTuple(const Tuple & r);
  void Verify(); // make sure the weights are in order. 
  void Build();
  void Store(string filename);
  void Load(string filename);
  Memblock StoredMemblock() {
    return Memblock((char*)data_, size_ * sizeof(Tuple));
  }
  void SumWeights(); // individual weights -> running sums
  void UnSumWeights(); // running sums -> individual weights
  static Sampler * Merge(Sampler *as1, Sampler *as2);
  uint64 WeightBefore(Tuple *p);
  
  // sample a tuple in the range [begin end)
  Tuple *Sample(const Tuple & begin, const Tuple & end, 
		RNG *rng, SampleStrategy strategy);
  uint64 RangeWeight(const Tuple & begin, const Tuple & end);
  uint64 RangeSize(const Tuple & begin, const Tuple & end);
  void CopyOutRangeLocked(const Tuple & begin, const Tuple &end,
			  uint64 **destination, uint64 * source_end_p);
  static void Test();

  // kth smallest weight per node using a sample of size n
  uint64 WeightPerNode(uint64 k, uint64 n);
};

inline Sampler* MakeEmptySampler() {
  Sampler * s = new Sampler;
  s->Allocate(0);
  s->Build();
  return s;
}

struct TupleJudge {
  // for a given tuple, what is the minimum weight at which it should be kept?
  virtual uint64 MinWeight(const Tuple & t) = 0;
  ~TupleJudge();
};
 
struct AdStore {
  AdStore();

  // -------------- READING ---------------
  // sample a tuple in the range [begin, end)
  int64 NumSamplers() { return samplers_.size(); }

  Tuple *Sample(const Tuple & begin, 
		const Tuple & end, 
		RWLock::Reader *reader, RNG *rng, SampleStrategy strategy);
  
  Tuple *Sample(const Tuple & prefix, RWLock::Reader *reader, 
		RNG *rng, SampleStrategy strategy);

  uint64 RangeWeight(const Tuple & begin, const Tuple & end, 
		     RWLock::Reader *reader);
  uint64 RangeWeight(const Tuple & prefix, RWLock::Reader *reader);

  uint64 RangeSizeLocked(const Tuple & begin, const Tuple &end);
  void CopyOutRangeLocked(const Tuple & begin, const Tuple &end,
			  uint64 **destination, uint64 * source_end_p);
  

  // After merging all the nonpermanents, you can call these
  Sampler *OnlyNonpermanentSampler() {
    CHECK(num_permanent_ + 1 == samplers_.size());
    return samplers_.back();
  }

  Sampler *OnlyPermanentSampler() {
    CHECK(num_permanent_ == 1);
    return samplers_[0];
  }
  
  // -------------- MODIFYING ---------------
  // called externally
  //  if force is set to false, may return false if busy. 
  bool AddSamplers(Sampler **start, uint64 num_samplers, bool force);

  // Make the existing samplers permanent.  
  // New samplers will never be merged into them.
  void MakePermanent();

  // ------------- MISC ----------------
  void Store(string filename);
  void Load(string filename);

  static void Test();

  // TODO: Garbage collection routines (dropping tuples)
  bool TooManyTuples();

  void Shrink();

  // --------------- PRIVATE INTERNALS ----------------

  void AddSamplerI(Sampler *s);

  // merge all non-permanent samplers.
  void MergeAllNonPermanentI();

  bool NeedsMerging();
  void MergeTopTwoSamplers();
  bool TooMuchTrash(); 
  void EmptyTrash();

  StableVector<Sampler *> samplers_;
  vector<Sampler *> trash_;
  uint64 num_permanent_;
  uint64 trash_tuples_;
  uint64 used_tuples_;
  uint64 max_tuples_;

  // External users need a Reader
  // Get a read lock if you want to sample. 
  // Get a write lock if you want to empty trash. 
  RWLock * sample_lock_;
  // Grab a write lock on this reader if you want to empty trash.
  // No two threads will try to use this reader at the same time, since 
  // anyone using it will already hold the modify_lock_.
  RWLock::Reader * sample_lock_reader_;

  // You grab this to modify. 
  pthread_mutex_t modify_lock_;
  
  static const uint64 merge_multiple_ = 4;  
};
CLOSEBRACE
